package com.lgCore.client.service.impl;

import com.lgCore.client.IZookServiceClient;
import com.lgCore.client.model.ClientZkData;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ZookClientImpl implements IZookServiceClient {
    private static final Log LOGGER = LogFactory.getLog(ZookClientImpl.class);
    private String basePath;

    private CuratorFramework zkClient;

    public ZookClientImpl(String basePath) {
        this.basePath ="/" + basePath;
    }

    @Override
    public void initServer() throws Exception {
        //获取全部节点信息
        List<String> childeList = zkClient.getChildren().forPath(basePath);
        if (childeList != null && !childeList.isEmpty()) {
            for (String servicePath : childeList) {
                List<String> childeList2 = zkClient.getChildren().forPath(basePath + "/" + servicePath);
                if (childeList2 != null && !childeList2.isEmpty()) {
                    for (String path : childeList2) {
                        String data = new String(zkClient.getData().forPath(basePath + "/" + servicePath + "/" + path));
                        ClientZkData.putData(servicePath, data);
                    }
                }
            }
        }
    }

    @Override
    public void monitorZkNode() throws Exception {
        ExecutorService executor = Executors.newCachedThreadPool();
        ThreadJob task = new ThreadJob();
        executor.submit(task);
    }

    @Override
    public void close() throws Exception {
        ClientZkData.clientZkDataHashMap.clear();
        zkClient.close();
    }

    @Override
    public void connectZookeeper(String server, int timeout) throws Exception {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000);
        zkClient = CuratorFrameworkFactory.builder()
                .connectString(server)
                .sessionTimeoutMs(timeout)
                .connectionTimeoutMs(timeout)
                .retryPolicy(retryPolicy)
                .build();
        zkClient.start();
    }

    private class ThreadJob implements Callable {

        @Override
        public Object call() throws Exception {

            TreeCache treeCache = new TreeCache(zkClient, basePath);
//        子节点删除, path=/serviceRpc/com.lgCore.service.UserService/192.168.10.157:10000000000001, data=192.168.10.157:1000
//        子节点增加, path=/serviceRpc/com.lgCore.service.UserService/192.168.10.157:10000000000002, data=192.168.10.157:1000
//        子节点更新, path=/serviceRpc/com.lgCore.service.UserService/192.168.10.157:10000000000002, data=192.168.10.157:10001
            //调用start方法开始监听
            treeCache.start();
            //添加TreeCacheListener监听器
            treeCache.getListenable().addListener(new TreeCacheListener() {
                @Override
                public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                    try {
                        ChildData data = event.getData();
                        String servicePath;
                        String dataPath;
                        String serviceName;
                        switch (event.getType()) {
                            case NODE_ADDED:

//                                LOGGER.info(String.format("子节点增加, path=%s, data=%s",
//                                        data.getPath(), new String(data.getData(), "UTF-8")));
                                servicePath = data.getPath();
                                dataPath = new String(data.getData(), "UTF-8");

                                serviceName = ClientZkData.getServicePath(servicePath);
                                if(serviceName!=null){
                                    // TODO: 2020/11/6  这里需要做调整，否则灰度发布有问题
                                    if("com.zjmy.crm.service.ClassService".equals(serviceName)){
                                        System.out.println(serviceName+","+dataPath);
                                    }
                                    ClientZkData.putData(serviceName, dataPath);
                                }
                                break;
                            case NODE_UPDATED:
//                                LOGGER.info(String.format("子节点更新, path=%s, data=%s",
//                                        data.getPath(), new String(data.getData(), "UTF-8")));
                                servicePath = data.getPath();
                                dataPath = new String(data.getData(), "UTF-8");
                                serviceName = ClientZkData.getServicePath(servicePath);
                                if(serviceName!=null){
                                    ClientZkData.editData(serviceName, dataPath);
                                }
                                break;
                            case NODE_REMOVED:
//                                LOGGER.info(String.format("子节点删除, path=%s, data=%s",
//                                        data.getPath(), new String(data.getData(), "UTF-8")));
                                servicePath = data.getPath();
                                dataPath = new String(data.getData(), "UTF-8");
                                serviceName = ClientZkData.getServicePath(servicePath);
                                if(serviceName!=null){
                                    ClientZkData.removeData(serviceName, dataPath);
                                }
                                break;
                            default:
                                break;
                        }

                    } catch (
                            UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }

                }
            });
            Thread.sleep(Integer.MAX_VALUE);
            return null;
        }
    }
}
